package com.shujia.transformation

import org.apache.flink.api.common.functions.FilterFunction
import org.apache.flink.streaming.api.scala._

object Demo03Filter {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    // 读取students.txt 过滤出文科班的学生
    val stuDS: DataStream[String] = env.readTextFile("Flink/data/stu/students.txt")
    // scala风格
    stuDS.filter(line => {
      line.split(",")(4).contains("文科")
    })
    //      .print()

    // java风格
    stuDS.filter(new FilterFunction[String] {
      override def filter(value: String): Boolean = {
        var flag = false
        val arr: Array[String] = value.split(",")
        if (arr.length == 5) {
          if (arr(4).contains("文科")) {
            flag = true
          }
        }
        flag
      }
    }).print()

    env.execute()


  }

}
